"""
任务管理器
"""
import json
from typing import Optional, List, Dict, Any
from datetime import datetime
from sqlalchemy import or_, cast, String, func, and_, desc
from .models import Task, CSGOKlineAnalysis
from .sqlalchemy_database import SQLAlchemyDatabaseManager
from com.constants.wear_ranges import WearCategory, WEAR_CATEGORY_RANGES, WEAR_CATEGORY_NAMES

class TaskManager:
    def __init__(self):
        """
        初始化任务管理器
        """
        self.db = SQLAlchemyDatabaseManager()

    def get_task_by_id(self, task_id: int) -> Optional[Dict[str, Any]]:
        """
        根据ID获取任务
        :param task_id: 任务ID
        :return: 任务信息字典或None
        """
        session = self.db.get_session()
        try:
            task = session.query(Task).filter(Task.id == task_id).first()
            return self._convert_model_to_dict(task) if task else None
        finally:
            session.close()

    def get_task_by_goods_id(self, goods_id: int) -> Optional[Dict[str, Any]]:
        """
        根据商品ID获取任务
        :param goods_id: 商品ID
        :return: 任务信息字典或None
        """
        session = self.db.get_session()
        try:
            task = session.query(Task).filter(Task.goods_id == goods_id).first()
            return self._convert_model_to_dict(task) if task else None
        finally:
            session.close()

    def get_all_tasks(self, limit: int = None, offset: int = None, keyword: str = None, goods_id: int = None, min_sell_num: int = None, min_price: float = None, max_price: float = None, status: int = None, wear_category: str = None, tags: list = None) -> List[Dict[str, Any]]:
        """
        获取所有任务
        :param limit: 限制返回数量
        :param offset: 偏移量（用于分页）
        :param keyword: 关键词
        :param goods_id: 商品ID
        :param min_sell_num: 最小在售数量
        :param min_price: 最小建议捡漏价
        :param max_price: 最大建议捡漏价
        :param status: 状态
        :param wear_category: 外观类别
        :param tags: 标签列表
        :return: 任务列表
        """
        session = self.db.get_session()
        
        # 创建子查询，获取每个商品的最新K线分析数据
        kline_subquery = session.query(
            CSGOKlineAnalysis.item_id,
            CSGOKlineAnalysis.timestamp,
            CSGOKlineAnalysis.overall_trend,
            CSGOKlineAnalysis.support_levels,
            CSGOKlineAnalysis.resistance_levels,
            CSGOKlineAnalysis.simple_advice,
            CSGOKlineAnalysis.analysis_details,
            func.row_number().over(
                partition_by=CSGOKlineAnalysis.item_id,
                order_by=desc(CSGOKlineAnalysis.timestamp)
            ).label('rn')
        ).subquery()
        
        # 筛选出每个商品的最新分析记录
        latest_kline = session.query(kline_subquery).filter(kline_subquery.c.rn == 1).subquery()
        
        # 构建查询
        query = session.query(
            Task,
            latest_kline.c.timestamp,
            latest_kline.c.overall_trend,
            latest_kline.c.support_levels,
            latest_kline.c.resistance_levels,
            latest_kline.c.simple_advice,
            latest_kline.c.analysis_details
        ).outerjoin(
            latest_kline,
            Task.goods_id == latest_kline.c.item_id
        )
        
        # 添加查询条件
        if keyword:
            query = query.filter(Task.name.like(f"%{keyword}%"))
        if goods_id:
            query = query.filter(Task.goods_id == int(goods_id))
        if min_sell_num:
            try:
                min_sell_num = int(min_sell_num)
                query = query.filter(Task.sell_num >= min_sell_num)
            except (ValueError, TypeError):
                pass
        if min_price:
            try:
                min_price = float(min_price)
                query = query.filter(Task.suggested_snipe_price >= min_price)
            except (ValueError, TypeError):
                pass
        if max_price:
            try:
                max_price = float(max_price)
                query = query.filter(Task.suggested_snipe_price <= max_price)
            except (ValueError, TypeError):
                pass
        if status is not None and status != -1:
            query = query.filter(Task.status == status)
        if wear_category:
            try:
                category = WearCategory(wear_category)
                # 使用中文名称进行查询
                chinese_name = WEAR_CATEGORY_NAMES[category]
                query = query.filter(Task.exterior == chinese_name)
            except (ValueError, KeyError):
                pass
        if tags and len(tags) > 0:
            for tag in tags:
                query = query.filter(Task.tags.contains(tag))
        
        # 排序和分页
        query = query.order_by(Task.create_at.desc())
        if limit is not None and offset is not None:
            query = query.limit(limit).offset(offset)
        
        # 执行查询并处理结果
        results = query.all()
        tasks = []
        for result in results:
            task_dict = self._convert_model_to_dict(result[0])  # 第一个元素是Task对象
            task_dict['kline_analysis'] = {
                'timestamp': result[1],
                'overall_trend': result[2],
                'support_levels': result[3],
                'resistance_levels': result[4],
                'simple_advice': result[5],
                'analysis_details': result[6]
            } if result[1] else None
            tasks.append(task_dict)
        
        return tasks

    def get_active_tasks(self) -> List[Dict[str, Any]]:
        """
        获取所有启用的任务
        :return: 任务列表
        """
        session = self.db.get_session()
        try:
            tasks = session.query(Task).filter(Task.status == 1).all()
            return [self._convert_model_to_dict(task) for task in tasks]
        finally:
            session.close()

    def search_tasks(self, keyword=None, goods_id=None, min_sell_num=None, min_price=None, max_price=None, status=None, wear_category=None, tags=None, limit=None, offset=None) -> List[Dict[str, Any]]:
        """
        搜索任务
        :param keyword: 关键词
        :param goods_id: 商品ID
        :param min_sell_num: 最小在售数量
        :param min_price: 最小建议捡漏价
        :param max_price: 最大建议捡漏价
        :param status: 状态
        :param wear_category: 外观类别
        :param tags: 标签列表
        :param limit: 限制数量
        :param offset: 偏移量
        :return: 任务列表
        """
        session, query = self.db.query_with_session(Task)
        try:
            # 构建子查询，获取每个任务的已购买数量
            from com.db.models import OrderRecord, ScanRecordPriceAnalysis, GoodsScoreAnalysis, CSGOKlineAnalysis
            
            # 购买数量子查询
            purchase_count_subquery = (
                session.query(
                    OrderRecord.task_id,
                    func.count(OrderRecord.id).label('purchase_count')
                )
                .filter(
                    OrderRecord.buy_status != 0,
                    OrderRecord.offer_status != 0
                )
                .group_by(OrderRecord.task_id)
                .subquery()
            )
            
            # 扫货记录分析子查询
            # 首先获取每个商品最新的分析时间
            latest_analysis_subquery = (
                session.query(
                    ScanRecordPriceAnalysis.goods_id,
                    func.max(ScanRecordPriceAnalysis.analysis_time).label('max_analysis_time')
                )
                .group_by(ScanRecordPriceAnalysis.goods_id)
                .subquery()
            )
            
            # 然后使用这个子查询来获取最新的分析记录
            price_analysis_subquery = (
                session.query(
                    ScanRecordPriceAnalysis.goods_id,
                    ScanRecordPriceAnalysis.analysis_time,
                    ScanRecordPriceAnalysis.abnormal_prices,
                    ScanRecordPriceAnalysis.common_sell_price,
                    ScanRecordPriceAnalysis.selling_count,
                    ScanRecordPriceAnalysis.normal_price,
                    ScanRecordPriceAnalysis.historical_low,
                    ScanRecordPriceAnalysis.price_distribution,
                    ScanRecordPriceAnalysis.abnormal_price_time_distribution,
                    ScanRecordPriceAnalysis.activity_time_distribution,
                    ScanRecordPriceAnalysis.record_count,
                    ScanRecordPriceAnalysis.top_price_list
                )
                .join(
                    latest_analysis_subquery,
                    and_(
                        ScanRecordPriceAnalysis.goods_id == latest_analysis_subquery.c.goods_id,
                        ScanRecordPriceAnalysis.analysis_time == latest_analysis_subquery.c.max_analysis_time
                    )
                )
                .subquery()
            )
            
            # 评分数据子查询
            score_subquery = (
                session.query(
                    GoodsScoreAnalysis.goods_id,
                    GoodsScoreAnalysis.total_score,
                    GoodsScoreAnalysis.grade,
                    GoodsScoreAnalysis.price_stability_score,
                    GoodsScoreAnalysis.profit_potential_score,
                    GoodsScoreAnalysis.trading_frequency_score,
                    GoodsScoreAnalysis.price_trend_score,
                    GoodsScoreAnalysis.time_activity_score,
                    GoodsScoreAnalysis.historical_success_score,
                    # 价格稳定性子评分
                    GoodsScoreAnalysis.stability_cv_score,
                    GoodsScoreAnalysis.stability_volatility_score,
                    # 利润空间子评分
                    GoodsScoreAnalysis.profit_rate_score,
                    GoodsScoreAnalysis.profit_frequency_score,
                    GoodsScoreAnalysis.profit_time_weighted_score,
                    # 交易频率子评分
                    GoodsScoreAnalysis.frequency_activity_score,
                    GoodsScoreAnalysis.frequency_selling_change_score,
                    GoodsScoreAnalysis.frequency_price_fluctuation_score,  # 添加价格波动得分
                    # 价格趋势子评分
                    GoodsScoreAnalysis.trend_direction_score,
                    GoodsScoreAnalysis.trend_range_score,
                    GoodsScoreAnalysis.trend_stability_score
                )
                .subquery()
            )
            
            # K线分析数据子查询
            # 首先获取每个商品最新的K线分析时间
            latest_kline_analysis_subquery = (
                session.query(
                    CSGOKlineAnalysis.item_id,
                    func.max(CSGOKlineAnalysis.timestamp).label('max_timestamp')
                )
                .group_by(CSGOKlineAnalysis.item_id)
                .subquery()
            )
            
            # 然后使用这个子查询来获取最新的K线分析记录
            kline_analysis_subquery = (
                session.query(
                    CSGOKlineAnalysis.item_id,
                    CSGOKlineAnalysis.timestamp,
                    CSGOKlineAnalysis.overall_trend,
                    CSGOKlineAnalysis.support_levels,
                    CSGOKlineAnalysis.resistance_levels,
                    CSGOKlineAnalysis.simple_advice
                )
                .join(
                    latest_kline_analysis_subquery,
                    and_(
                        CSGOKlineAnalysis.item_id == latest_kline_analysis_subquery.c.item_id,
                        CSGOKlineAnalysis.timestamp == latest_kline_analysis_subquery.c.max_timestamp
                    )
                )
                .subquery()
            )
            
            # 将子查询添加到主查询中
            query = query.outerjoin(
                purchase_count_subquery,
                Task.id == purchase_count_subquery.c.task_id
            ).outerjoin(
                price_analysis_subquery,
                Task.goods_id == price_analysis_subquery.c.goods_id
            ).outerjoin(
                score_subquery,
                Task.goods_id == score_subquery.c.goods_id
            ).outerjoin(
                kline_analysis_subquery,
                cast(Task.goods_id, String) == kline_analysis_subquery.c.item_id
            )
            
            # 添加已购买数量、扫货分析数据和评分数据到查询结果中
            query = query.add_columns(
                func.coalesce(purchase_count_subquery.c.purchase_count, 0).label('purchase_count'),
                price_analysis_subquery.c.analysis_time.label('analysis_time'),
                price_analysis_subquery.c.abnormal_prices.label('abnormal_prices'),
                price_analysis_subquery.c.common_sell_price.label('common_sell_price'),
                price_analysis_subquery.c.selling_count.label('analysis_selling_count'),
                price_analysis_subquery.c.normal_price.label('normal_price'),
                price_analysis_subquery.c.historical_low.label('historical_low'),
                price_analysis_subquery.c.price_distribution.label('price_distribution'),
                price_analysis_subquery.c.abnormal_price_time_distribution.label('abnormal_price_time_distribution'),
                price_analysis_subquery.c.activity_time_distribution.label('activity_time_distribution'),
                price_analysis_subquery.c.record_count.label('record_count'),
                price_analysis_subquery.c.top_price_list.label('top_price_list'),
                score_subquery.c.total_score,
                score_subquery.c.grade,
                score_subquery.c.price_stability_score,
                score_subquery.c.profit_potential_score,
                score_subquery.c.trading_frequency_score,
                score_subquery.c.price_trend_score,
                score_subquery.c.time_activity_score,
                score_subquery.c.historical_success_score,
                score_subquery.c.stability_cv_score,
                score_subquery.c.stability_volatility_score,
                score_subquery.c.profit_rate_score,
                score_subquery.c.profit_frequency_score,
                score_subquery.c.profit_time_weighted_score,
                score_subquery.c.frequency_activity_score,
                score_subquery.c.frequency_selling_change_score,
                score_subquery.c.frequency_price_fluctuation_score,  # 添加价格波动得分
                score_subquery.c.trend_direction_score,  # 趋势-方向和强度得分
                score_subquery.c.trend_range_score,  # 趋势-价格变化幅度得分
                score_subquery.c.trend_stability_score,  # 趋势-稳定性得分
                kline_analysis_subquery.c.timestamp.label('kline_timestamp'),
                kline_analysis_subquery.c.overall_trend.label('kline_overall_trend'),
                kline_analysis_subquery.c.support_levels.label('kline_support_levels'),
                kline_analysis_subquery.c.resistance_levels.label('kline_resistance_levels'),
                kline_analysis_subquery.c.simple_advice.label('kline_simple_advice')
            )
            
            # 添加查询条件
            if keyword:
                query = query.filter(Task.name.like(f"%{keyword}%"))
            if goods_id:
                query = query.filter(Task.goods_id == int(goods_id))
            if min_sell_num:
                try:
                    min_sell_num = int(min_sell_num)
                    query = query.filter(Task.sell_num >= min_sell_num)
                except (ValueError, TypeError):
                    pass
            if min_price:
                try:
                    min_price = float(min_price)
                    query = query.filter(Task.suggested_snipe_price >= min_price)
                except (ValueError, TypeError):
                    pass
            if max_price:
                try:
                    max_price = float(max_price)
                    query = query.filter(Task.suggested_snipe_price <= max_price)
                except (ValueError, TypeError):
                    pass
            if status is not None and status != -1:
                query = query.filter(Task.status == status)
            if wear_category:
                try:
                    category = WearCategory(wear_category)
                    # 使用中文名称进行查询
                    chinese_name = WEAR_CATEGORY_NAMES[category]
                    query = query.filter(Task.exterior == chinese_name)
                except (ValueError, KeyError):
                    pass
            if tags and len(tags) > 0:
                for tag in tags:
                    query = query.filter(Task.tags.contains(tag))
            
            # 排序和分页
            query = query.order_by(Task.create_at.desc())
            if limit is not None and offset is not None:
                query = query.limit(limit).offset(offset)
            
            # 执行查询并处理结果
            results = query.all()
            tasks = []
            for result in results:
                task_dict = self._convert_model_to_dict(result[0])  # 第一个元素是Task对象
                task_dict['purchase_count'] = result.purchase_count  # 添加已购买数量
                
                # 添加扫货分析数据
                task_dict['price_analysis'] = {
                    'analysis_time': result.analysis_time,
                    'abnormal_prices': result.abnormal_prices,
                    'common_sell_price': result.common_sell_price,
                    'selling_count': result.analysis_selling_count,
                    'normal_price': result.normal_price,
                    'historical_low': result.historical_low,
                    'price_distribution': result.price_distribution,
                    'abnormal_price_time_distribution': result.abnormal_price_time_distribution,
                    'activity_time_distribution': result.activity_time_distribution,
                    'record_count': result.record_count,
                    'top_price_list': result.top_price_list
                } if result.analysis_time else None
                
                # 添加评分数据
                task_dict['dscore'] = {
                    'total_score': result.total_score,
                    'grade': result.grade,
                    'price_stability_score': result.price_stability_score,
                    'profit_potential_score': result.profit_potential_score,
                    'trading_frequency_score': result.trading_frequency_score,
                    'price_trend_score': result.price_trend_score,
                    'time_activity_score': result.time_activity_score,
                    'historical_success_score': result.historical_success_score,
                    'stability_cv_score': result.stability_cv_score,
                    'stability_volatility_score': result.stability_volatility_score,
                    'profit_rate_score': result.profit_rate_score,
                    'profit_frequency_score': result.profit_frequency_score,
                    'profit_time_weighted_score': result.profit_time_weighted_score,
                    'frequency_activity_score': result.frequency_activity_score,
                    'frequency_selling_change_score': result.frequency_selling_change_score,
                    'frequency_price_fluctuation_score': result.frequency_price_fluctuation_score,  # 添加价格波动得分
                    'trend_direction_score': result.trend_direction_score,  # 趋势-方向和强度得分
                    'trend_range_score': result.trend_range_score,  # 趋势-价格变化幅度得分
                    'trend_stability_score': result.trend_stability_score  # 趋势-稳定性得分
                } if result.total_score else None
                
                # 添加K线分析数据
                task_dict['kline_analysis'] = {
                    'timestamp': result.kline_timestamp,
                    'overall_trend': result.kline_overall_trend,
                    'support_levels': result.kline_support_levels,
                    'resistance_levels': result.kline_resistance_levels,
                    'simple_advice': result.kline_simple_advice
                } if result.kline_timestamp else None
                
                tasks.append(task_dict)
            
            return tasks
        finally:
            session.close()

    def get_task_count(self, keyword=None, goods_id=None, min_sell_num=None, min_price=None, max_price=None, status=None, wear_category=None, tags=None) -> int:
        """获取任务总数"""
        session, query = self.db.query_with_session(Task)
        try:
            if keyword:
                query = query.filter(Task.name.like(f"%{keyword}%"))
            if goods_id:
                query = query.filter(Task.goods_id == int(goods_id))
            if min_sell_num:
                try:
                    min_sell_num = int(min_sell_num)
                    query = query.filter(Task.sell_num >= min_sell_num)
                except (ValueError, TypeError):
                    pass
            if min_price:
                try:
                    min_price = float(min_price)
                    query = query.filter(Task.suggested_snipe_price >= min_price)
                except (ValueError, TypeError):
                    pass
            if max_price:
                try:
                    max_price = float(max_price)
                    query = query.filter(Task.suggested_snipe_price <= max_price)
                except (ValueError, TypeError):
                    pass
            if status is not None and status != -1:
                query = query.filter(Task.status == status)
            if wear_category:
                try:
                    category = WearCategory(wear_category)
                    # 使用中文名称进行查询
                    chinese_name = WEAR_CATEGORY_NAMES[category]
                    query = query.filter(Task.exterior == chinese_name)
                except (ValueError, KeyError):
                    pass
            if tags and len(tags) > 0:
                for tag in tags:
                    query = query.filter(Task.tags.contains(tag))
            
            return query.count()
        finally:
            session.close()

    def get_disabled_tasks(self) -> List[Dict[str, Any]]:
        """
        获取所有禁用状态的任务
        :return: 任务列表
        """
        session = self.db.get_session()
        try:
            tasks = session.query(Task).filter(Task.disabled == 1).all()
            return [self._convert_model_to_dict(task) for task in tasks]
        finally:
            session.close()

    def add_task(self, task_data: Dict[str, Any]) -> int:
        """
        添加新任务
        :param task_data: 任务数据
        :return: 新任务的ID
        """
        task_data = self._prepare_task_data(task_data)
        task = Task(**task_data)
        task_id = self.db.add(task)  # 直接使用返回的 ID
        return task_id

    def update_task(self, task_id: int, buy_price_limit: float = None, buy_num_limit: int = None,
                   min_paintwear: float = None, max_paintwear: float = None, tags: list = None,
                   status: int = None, pricing_type: int = None, profit_rate: float = None, min_profit_rate: float = None) -> bool:
        """
        更新任务
        :param task_id: 任务ID
        :param buy_price_limit: 最大购买价格
        :param buy_num_limit: 购买数量限制
        :param min_paintwear: 最小磨损度
        :param max_paintwear: 最大磨损度
        :param tags: 标签列表
        :param status: 状态（1：启用，0：停用）
        :param pricing_type: 扫货定价方式类型（0：固定价格，1：动态利润率，2：固定利润率）
        :param profit_rate: 利润率
        :param min_profit_rate: 最低利润率
        :return: 是否更新成功
        """
        try:
            session = self.db.get_session()
            try:
                task = session.get(Task, task_id)
                if not task:
                    return False
                    
                # 更新购买价格限制
                if buy_price_limit is not None:
                    task.buy_price_limit = buy_price_limit
                    
                # 更新购买数量限制
                if buy_num_limit is not None:
                    task.buy_num_limit = buy_num_limit
                    
                # 更新磨损度范围
                if min_paintwear is not None or max_paintwear is not None:
                    filter_conf = json.loads(task.filter_conf) if task.filter_conf else {}
                    if min_paintwear is not None:
                        filter_conf['min_paintwear'] = str(min_paintwear)
                    if max_paintwear is not None:
                        filter_conf['max_paintwear'] = str(max_paintwear)
                    filter_conf['type'] = 'paintwear'
                    task.filter_conf = json.dumps(filter_conf)
                    
                # 更新标签 - 保存为 JSON 格式
                if tags is not None:
                    task.tags = json.dumps(tags) if tags else '[]'
                    
                # 更新状态
                if status is not None:
                    task.status = status
                
                # 更新扫货定价方式类型
                if pricing_type is not None:
                    task.pricing_type = pricing_type
                
                # 更新利润率
                if profit_rate is not None:
                    task.profit_rate = profit_rate
                
                # 更新最低利润率
                if min_profit_rate is not None:
                    task.min_profit_rate = min_profit_rate
                    
                # 更新时间戳
                task.update_at = int(datetime.now().timestamp() * 1000)
                
                session.commit()
                return True
            finally:
                session.close()
        except Exception as e:
            print(f"更新任务失败: {str(e)}")
            return False

    def update_task_status(self, task_id: int, status: int) -> bool:
        """
        更新任务状态
        :param task_id: 任务ID
        :param status: 状态（1：启用，0：停用）
        :return: 是否更新成功
        """
        try:
            task = self.db.query(Task).filter(Task.id == task_id).first()
            if task:
                task.status = status
                task.update_at = int(datetime.now().timestamp() * 1000)
                self.db.update_obj(task)
                return True
            return False
        except Exception as e:
            return False

    def update_task_prices(self, task_id: int, buy_price_limit: float = None, target_sell_price: float = None) -> bool:
        """
        更新任务的价格相关字段
        
        Args:
            task_id: 任务ID
            buy_price_limit: 买入价格限制
            target_sell_price: 目标售价
            
        Returns:
            bool: 是否更新成功
        """
        try:
            session = self.db.get_session()
            try:
                task = session.query(Task).filter(Task.id == task_id).first()
                if not task:
                    print(f"未找到任务: {task_id}")
                    return False
                    
                if buy_price_limit is not None:
                    task.buy_price_limit = buy_price_limit
                    print(f"更新任务买入价格限制: {task_id} -> {buy_price_limit}")
                    
                if target_sell_price is not None:
                    task.target_sell_price = target_sell_price
                    print(f"更新任务目标售价: {task_id} -> {target_sell_price}")
                    
                task.update_at = int(datetime.now().timestamp() * 1000)
                session.commit()
                print(f"成功更新任务价格: {task_id}")
                return True
            finally:
                session.close()
        except Exception as e:
            print(f"更新任务价格失败: {str(e)}")
            return False

    def update_task_profit_rate(self, task_id: int, profit_rate: float) -> bool:
        """
        更新任务的利润率字段
        
        Args:
            task_id: 任务ID
            profit_rate: 利润率（百分比值，如 5.0 表示 5%）
            
        Returns:
            bool: 是否更新成功
        """
        try:
            session = self.db.get_session()
            try:
                task = session.query(Task).filter(Task.id == task_id).first()
                if not task:
                    print(f"未找到任务: {task_id}")
                    return False
                
                # 更新利润率字段
                task.profit_rate = profit_rate
                print(f"更新任务利润率: {task_id} -> {profit_rate}%")
                
                task.update_at = int(datetime.now().timestamp() * 1000)
                session.commit()
                print(f"成功更新任务利润率: {task_id}")
                return True
            finally:
                session.close()
        except Exception as e:
            print(f"更新任务利润率失败: {str(e)}")
            return False

    def update_task_min_profit_rate(self, task_id: int, min_profit_rate: float) -> bool:
        """
        更新任务的最低利润率字段
        
        Args:
            task_id: 任务ID
            min_profit_rate: 最低利润率（百分比值，如 3.0 表示 3%）
            
        Returns:
            bool: 是否更新成功
        """
        try:
            session = self.db.get_session()
            try:
                task = session.query(Task).filter(Task.id == task_id).first()
                if not task:
                    print(f"未找到任务: {task_id}")
                    return False
                
                # 更新最低利润率字段
                task.min_profit_rate = min_profit_rate
                print(f"更新任务最低利润率: {task_id} -> {min_profit_rate}%")
                
                task.update_at = int(datetime.now().timestamp() * 1000)
                session.commit()
                print(f"成功更新任务最低利润率: {task_id}")
                return True
            finally:
                session.close()
        except Exception as e:
            print(f"更新任务最低利润率失败: {str(e)}")
            return False

    def update_bought_num(self, task_id: int, increment: int = 1) -> bool:
        """
        更新已购买数量
        :param task_id: 任务ID
        :param increment: 增加数量
        :return: 是否成功
        """
        task = self.db.get_by_id(Task, task_id)
        if not task:
            return False
            
        task.bought_num += increment
        task.update_at = int(datetime.now().timestamp() * 1000)
        self.db.update_obj(task)
        return True

    def disable_task(self, task_id: int, reason: str) -> bool:
        """
        禁用任务
        :param task_id: 任务ID
        :param reason: 禁用原因
        :return: 是否成功
        """
        task = self.db.get_by_id(Task, task_id)
        if not task:
            return False
            
        now = int(datetime.now().timestamp() * 1000)
        task.disabled = 1
        task.disabled_desc = reason
        task.disabled_at = now
        task.update_at = now
        
        self.db.update_obj(task)
        return True

    def enable_task(self, task_id: int) -> bool:
        """
        启用任务
        :param task_id: 任务ID
        :return: 是否成功
        """
        task = self.db.get_by_id(Task, task_id)
        if not task:
            return False
            
        now = int(datetime.now().timestamp() * 1000)
        task.disabled = 0
        task.disabled_desc = ''
        task.disabled_at = 0
        task.disabled_until = 0
        task.update_at = now
        
        self.db.update_obj(task)
        return True

    def delete_task(self, task_id: int) -> None:
        """
        删除指定ID的任务
        :param task_id: 任务ID
        """
        try:
            # 查找任务
            task = self.db.get_by_id(Task, task_id)
            if not task:
                raise ValueError(f"任务ID {task_id} 不存在")
                
            # 删除任务
            self.db.delete_obj(task)
        except Exception as e:
            raise Exception(f"删除任务失败: {str(e)}")

    def get_tasks_by_script_id(self, script_id: int) -> List[Dict[str, Any]]:
        """
        获取指定脚本ID关联的所有任务
        :param script_id: 脚本ID
        :return: 任务列表
        """
        session = self.db.get_session()
        try:
            tasks = session.query(Task)\
                .join("script_task", Task.id == "task_id")\
                .filter(Task.status == 1)\
                .filter("script_id == :script_id")\
                .params(script_id=script_id)\
                .all()
            return [self._convert_model_to_dict(task) for task in tasks]
        finally:
            session.close()

    def get_all_tags(self) -> List[str]:
        """
        获取所有任务中使用的标签
        :return: 标签列表
        """
        session = self.db.get_session()
        try:
            tasks = session.query(Task).filter(Task.tags != '[]').all()
            all_tags = set()
            
            for task in tasks:
                try:
                    tags = json.loads(task.tags)
                    all_tags.update(tags)
                except (json.JSONDecodeError, TypeError):
                    continue
                
            return sorted(list(all_tags))
        finally:
            session.close()

    def _prepare_task_data(self, task_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        准备任务数据，设置默认值和处理特殊字段
        :param task_data: 原始任务数据
        :return: 处理后的任务数据
        """
        # 创建数据副本，避免修改原始数据
        data = task_data.copy()
        
        # 处理时间戳
        now = int(datetime.now().timestamp() * 1000)
        data['update_at'] = now
        if 'create_at' not in data:
            data['create_at'] = str(now)  # create_at 是 String 类型
        elif isinstance(data['create_at'], int):
            data['create_at'] = str(data['create_at'])  # 转换为字符串
            
        # 设置默认值
        data.setdefault('status', 1)  # 默认启用
        data.setdefault('update_price_auto', 0)  # 默认不自动更新
        data.setdefault('bought_num', 0)  # 默认已购数量为0
        data.setdefault('buying_num', 0)  # 默认正在购买数量为0
        data.setdefault('disabled', 0)  # 默认不禁用
        data.setdefault('disabled_desc', '')  # 默认禁用描述为空
        data.setdefault('disabled_at', 0)  # 默认禁用时间为0
        data.setdefault('disabled_until', 0)  # 默认禁用截止时间为0
        
        # 处理标签字段
        if 'tags' in data:
            if isinstance(data['tags'], list):
                data['tags'] = json.dumps(data['tags'])
            elif data['tags'] is None:
                data['tags'] = '[]'
        else:
            data['tags'] = '[]'
        
        # 处理数值字段，确保是浮点数或整数
        float_fields = ['sell_min_price', 'sell_max_price', 'buy_price_limit', 'filter_min_price', 
                       'suggested_snipe_price', 'theoretical_max_buy_price', 'target_sell_price', 
                       'score', 'liquidity_score', 'price_advantage_score', 'stability_score', 
                       'supply_score', 'daily_volume', 'price_range_compactness', 'volatility']
        for field in float_fields:
            if field in data:
                try:
                    data[field] = float(data[field]) if data[field] is not None else None
                except (ValueError, TypeError):
                    data[field] = None
        
        int_fields = ['buy_num_limit', 'min_interval', 'sell_num', 'goods_task_num']
        for field in int_fields:
            if field in data:
                try:
                    data[field] = int(data[field]) if data[field] is not None else None
                except (ValueError, TypeError):
                    data[field] = None
        
        # 处理 JSON 配置字段
        json_fields = ['filter_conf', 'update_price_conf']
        for field in json_fields:
            if field in data:
                if isinstance(data[field], dict):
                    data[field] = json.dumps(data[field])
                elif data[field] is None:
                    data[field] = '{}'
        
        return data

    def _convert_model_to_dict(self, task: Task) -> Dict[str, Any]:
        """
        将Task模型转换为字典，并处理JSON字段的反序列化
        """
        if not task:
            return {}
            
        result = {
            'id': task.id,
            'goods_id': task.goods_id,
            'name': task.name,
            'hash_name': task.hash_name,
            'wear': task.wear,
            'rarity': task.rarity,
            'quality': task.quality,
            'exterior': task.exterior,
            'score': task.score,
            'liquidity_score': task.liquidity_score,
            'price_advantage_score': task.price_advantage_score,
            'stability_score': task.stability_score,
            'supply_score': task.supply_score,
            'daily_volume': task.daily_volume,
            'sell_min_price': task.sell_min_price,
            'sell_max_price': task.sell_max_price,
            'suggested_snipe_price': task.suggested_snipe_price,
            'target_sell_price': task.target_sell_price,
            'theoretical_max_buy_price': task.theoretical_max_buy_price,
            'historical_price_range': task.historical_price_range,
            'trend_analysis': task.trend_analysis,  # 添加价格趋势字段
            'price_range_compactness': task.price_range_compactness,  # 添加价格紧凑度字段
            'volatility': task.volatility,  # 添加波动率字段
            'tags': task.tags,
            'filter_conf': task.filter_conf,
            'buy_price_limit': task.buy_price_limit,
            'update_price_auto': task.update_price_auto,
            'update_price_conf': task.update_price_conf,
            'buy_num_limit': task.buy_num_limit,
            'bought_num': task.bought_num,
            'min_interval': task.min_interval,
            'status': task.status,
            'goods_url': task.goods_url,
            'sell_num': task.sell_num,
            'filter_min_price': task.filter_min_price,
            'create_at': task.create_at,
            'update_at': task.update_at,
            'buying_num': task.buying_num,
            'disabled': task.disabled,
            'disabled_desc': task.disabled_desc,
            'disabled_at': task.disabled_at,
            'disabled_until': task.disabled_until,
            'goods_task_num': task.goods_task_num,
            'goods_task_status': task.goods_task_status,
            'pricing_type': task.pricing_type,
            'profit_rate': task.profit_rate,
            'min_profit_rate': task.min_profit_rate
        }
        
        try:
            if result['tags']:
                result['tags'] = json.loads(result['tags'])
            if result['filter_conf']:
                result['filter_conf'] = json.loads(result['filter_conf'])
            if result['update_price_conf']:
                result['update_price_conf'] = json.loads(result['update_price_conf'])
            if result['historical_price_range']:
                result['historical_price_range'] = json.loads(result['historical_price_range'])
            if result['trend_analysis']:  # 添加对价格趋势的JSON解析
                result['trend_analysis'] = json.loads(result['trend_analysis'])
        except (json.JSONDecodeError, TypeError):
            pass  # 如果JSON解析失败，保持原始字符串
            
        return result
